package top.reminis.utils;

import lombok.extern.slf4j.Slf4j;
import top.reminis.conf.JobConfig;

import java.util.*;

/**
 * 配置驱动的SQL生成器
 * 支持从JobConfig动态生成SQL语句
 *
 * @author zwf
 * @create 2025-09-24
 */
@Slf4j
public class ConfigurableSqlGenerator {

    private static JobConfig jobConfig;

    /**
     * 设置作业配置
     */
    public static void setJobConfig(JobConfig config) {
        jobConfig = config;
        log.info("ConfigurableSqlGenerator 已设置作业配置: {}", config.getJobName());
    }

    /**
     * 检查配置是否已设置
     */
    private static void checkConfig() {
        if (jobConfig == null) {
            throw new IllegalStateException("ConfigurableSqlGenerator 尚未设置作业配置，请先调用 setJobConfig()");
        }
    }

    /**
     * 生成Kafka源表SQL
     */
    public static String generateKafkaSourceTable(JobConfig.DataSourceConfig sourceConfig) {
        checkConfig();

        StringBuilder sql = new StringBuilder();
        sql.append("CREATE TEMPORARY TABLE ").append(sourceConfig.getName()).append(" (\n");

        // 使用配置文件中的字段定义
        if (sourceConfig.getFields() != null && !sourceConfig.getFields().isEmpty()) {
            for (int i = 0; i < sourceConfig.getFields().size(); i++) {
                JobConfig.FieldDefinition field = sourceConfig.getFields().get(i);
                sql.append("  ").append(field.getName()).append(" ").append(field.getType());
                
                // 添加注释
                if (field.getComment() != null && !field.getComment().isEmpty()) {
                    sql.append(" COMMENT '").append(field.getComment()).append("'");
                }
                
                // 添加非空约束
                if (field.isNotNull()) {
                    sql.append(" NOT NULL");
                }
                
                // 添加默认值
                if (field.getDefaultValue() != null && !field.getDefaultValue().isEmpty()) {
                    sql.append(" DEFAULT ").append(field.getDefaultValue());
                }
                
                // 如果不是最后一个字段，添加逗号
                if (i < sourceConfig.getFields().size() - 1) {
                    sql.append(",");
                }
                sql.append("\n");
            }
        } else {
            // 如果没有字段定义，抛出异常终止程序
            throw new IllegalStateException("数据源 " + sourceConfig.getName() + " 没有字段定义，请在配置文件中添加 fields 配置");
        }

        sql.append(") WITH (\n");
        sql.append("  'connector' = '").append(sourceConfig.getType()).append("',\n");
        sql.append("  'topic' = '").append(sourceConfig.getTopic()).append("',\n");
        sql.append("  'properties.bootstrap.servers' = '").append(sourceConfig.getBootstrapServers()).append("',\n");
        sql.append("  'properties.group.id' = '").append(sourceConfig.getGroupId()).append("',\n");

        // 根据配置文件中的属性生成连接属性
        if (sourceConfig.getProperties() != null) {
            for (Map.Entry<String, String> entry : sourceConfig.getProperties().entrySet()) {
                sql.append("  '").append(entry.getKey()).append("' = '").append(entry.getValue()).append("',\n");
            }
        }

        // 移除最后一个逗号
        if (sql.toString().endsWith(",\n")) {
            sql.setLength(sql.length() - 2);
            sql.append("\n");
        }

        sql.append(")");

        return sql.toString();
    }

    /**
     * 生成Iceberg目标表SQL
     */
    public static String generateIcebergSinkTable(JobConfig.DataSinkConfig sinkConfig) {
        checkConfig();

        // 检查是否是 Kafka 死信队列
        if ("kafka".equalsIgnoreCase(sinkConfig.getType())) {
            return generateKafkaDlqTable(sinkConfig);
        }

        StringBuilder sql = new StringBuilder();
        sql.append("CREATE TABLE ").append(sinkConfig.getTableName()).append(" (\n");

        // 使用配置文件中的字段定义
        List<String> primaryKeyFields = new ArrayList<>();
        
        if (sinkConfig.getFields() != null && !sinkConfig.getFields().isEmpty()) {
            for (int i = 0; i < sinkConfig.getFields().size(); i++) {
                JobConfig.FieldDefinition field = sinkConfig.getFields().get(i);
                sql.append("  ").append(field.getName()).append(" ").append(field.getType());
                
                // 添加注释
                if (field.getComment() != null && !field.getComment().isEmpty()) {
                    sql.append(" COMMENT '").append(field.getComment()).append("'");
                }
                
                // 添加默认值
                if (field.getDefaultValue() != null && !field.getDefaultValue().isEmpty()) {
                    sql.append(" DEFAULT ").append(field.getDefaultValue());
                }
                
                // 注意：Flink SQL 中的 Iceberg 表不支持 NOT NULL 约束
                // 非空约束通过 Iceberg 表的 schema 来管理
                
                // 收集主键字段
                if (field.isPrimaryKey()) {
                    primaryKeyFields.add(field.getName());
                }
                
                // 如果不是最后一个字段，添加逗号
                if (i < sinkConfig.getFields().size() - 1) {
                    sql.append(",");
                }
                sql.append("\n");
            }
            
        } else {
            // 如果没有字段定义，抛出异常终止程序
            throw new IllegalStateException("数据目标 " + sinkConfig.getName() + " 没有字段定义，请在配置文件中添加 fields 配置");
        }
        
        // 添加主键约束（在字段定义之后）
        if (!primaryKeyFields.isEmpty()) {
            sql.append(", PRIMARY KEY (").append(String.join(", ", primaryKeyFields)).append(") NOT ENFORCED\n");
        }

        sql.append(") PARTITIONED BY (dt)\n");

        // 添加表属性
        sql.append("WITH (\n");
        if (sinkConfig.getProperties() != null) {
            for (Map.Entry<String, String> entry : sinkConfig.getProperties().entrySet()) {
                sql.append("  '").append(entry.getKey()).append("' = '").append(entry.getValue()).append("',\n");
            }
        }

        // 移除最后一个逗号
        if (sql.toString().endsWith(",\n")) {
            sql.setLength(sql.length() - 2);
            sql.append("\n");
        }

        sql.append(")");

        return sql.toString();
    }

    /**
     * 生成Kafka死信队列表SQL
     */
    private static String generateKafkaDlqTable(JobConfig.DataSinkConfig sinkConfig) {
        // 验证字段定义配置
        if (sinkConfig.getFields() == null || sinkConfig.getFields().isEmpty()) {
            throw new IllegalStateException("Kafka死信队列表 " + sinkConfig.getName() + " 缺少字段定义配置，无法生成表SQL");
        }

        StringBuilder sql = new StringBuilder();
        sql.append("CREATE TEMPORARY TABLE ").append(sinkConfig.getName()).append(" (\n");

        // 使用配置文件中的字段定义
        for (int i = 0; i < sinkConfig.getFields().size(); i++) {
            JobConfig.FieldDefinition field = sinkConfig.getFields().get(i);
            sql.append("  ").append(field.getName()).append(" ").append(field.getType());
            
            // 添加注释
            if (field.getComment() != null && !field.getComment().isEmpty()) {
                sql.append(" COMMENT '").append(field.getComment()).append("'");
            }
            
            // 添加默认值
            if (field.getDefaultValue() != null && !field.getDefaultValue().isEmpty()) {
                sql.append(" DEFAULT ").append(field.getDefaultValue());
            }
            
            // 如果不是最后一个字段，添加逗号
            if (i < sinkConfig.getFields().size() - 1) {
                sql.append(",");
            }
            sql.append("\n");
        }

        sql.append(") WITH (\n");
        sql.append("  'connector' = 'kafka',\n");
        sql.append("  'topic' = '").append(sinkConfig.getTopic()).append("',\n");
        sql.append("  'properties.bootstrap.servers' = '").append(sinkConfig.getBootstrapServers()).append("',\n");

        // 添加属性
        if (sinkConfig.getProperties() != null) {
            for (Map.Entry<String, String> entry : sinkConfig.getProperties().entrySet()) {
                sql.append("  '").append(entry.getKey()).append("' = '").append(entry.getValue()).append("',\n");
            }
        }

        // 移除最后一个逗号
        if (sql.toString().endsWith(",\n")) {
            sql.setLength(sql.length() - 2);
            sql.append("\n");
        }

        sql.append(")");

        return sql.toString();
    }

    /**
     * 生成数据转换SQL
     */
    public static String generateDataTransformationSql(JobConfig.DataFlowConfig dataFlow) {
        checkConfig();

        // 验证字段映射配置
        if (dataFlow.getFieldMappings() == null || dataFlow.getFieldMappings().isEmpty()) {
            throw new IllegalStateException("数据流 " + dataFlow.getName() + " 缺少字段映射配置，无法生成转换SQL");
        }

        StringBuilder sql = new StringBuilder();
        sql.append("INSERT INTO ").append(dataFlow.getTargetTable()).append("\n");
        sql.append("SELECT\n");

        // 使用配置文件中的字段映射
        for (int i = 0; i < dataFlow.getFieldMappings().size(); i++) {
            JobConfig.FieldMapping mapping = dataFlow.getFieldMappings().get(i);
            
            // 验证必要字段
            if (mapping.getSource() == null || mapping.getTarget() == null) {
                throw new IllegalStateException("字段映射配置不完整: source=" + mapping.getSource() + ", target=" + mapping.getTarget());
            }
            
            // 生成字段映射
            if (mapping.getTransform() != null && !mapping.getTransform().isEmpty()) {
                // 使用自定义转换表达式
                sql.append("  ").append(mapping.getTransform()).append(" AS ").append(mapping.getTarget());
            } else {
                // 直接映射
                sql.append("  ").append(mapping.getSource()).append(" AS ").append(mapping.getTarget());
            }
            
            // 如果不是最后一个字段，添加逗号
            if (i < dataFlow.getFieldMappings().size() - 1) {
                sql.append(",");
            }
            sql.append("\n");
        }

        sql.append("FROM ").append(dataFlow.getSourceTable()).append("\n");

        // 添加过滤条件
        if (dataFlow.getFilterCondition() != null && !dataFlow.getFilterCondition().isEmpty()) {
            sql.append("WHERE ").append(dataFlow.getFilterCondition());
        }

        return sql.toString();
    }

    /**
     * 生成死信队列SQL
     */
    public static String generateDlqSql(JobConfig.DataFlowConfig dataFlow) {
        checkConfig();

        // 验证字段映射配置
        if (dataFlow.getFieldMappings() == null || dataFlow.getFieldMappings().isEmpty()) {
            throw new IllegalStateException("死信队列数据流 " + dataFlow.getName() + " 缺少字段映射配置，无法生成DLQ SQL");
        }

        StringBuilder sql = new StringBuilder();
        sql.append("INSERT INTO ").append(dataFlow.getTargetTable()).append("\n");
        sql.append("SELECT\n");

        // 使用配置文件中的字段映射
        for (int i = 0; i < dataFlow.getFieldMappings().size(); i++) {
            JobConfig.FieldMapping mapping = dataFlow.getFieldMappings().get(i);
            
            // 验证必要字段
            if (mapping.getSource() == null || mapping.getTarget() == null) {
                throw new IllegalStateException("DLQ字段映射配置不完整: source=" + mapping.getSource() + ", target=" + mapping.getTarget());
            }
            
            // 生成字段映射
            if (mapping.getTransform() != null && !mapping.getTransform().isEmpty()) {
                // 使用自定义转换表达式
                sql.append("  ").append(mapping.getTransform()).append(" AS ").append(mapping.getTarget());
            } else {
                // 直接映射
                sql.append("  ").append(mapping.getSource()).append(" AS ").append(mapping.getTarget());
            }
            
            // 如果不是最后一个字段，添加逗号
            if (i < dataFlow.getFieldMappings().size() - 1) {
                sql.append(",");
            }
            sql.append("\n");
        }

        sql.append("FROM ").append(dataFlow.getSourceTable()).append("\n");

        // 添加过滤条件
        if (dataFlow.getFilterCondition() != null && !dataFlow.getFilterCondition().isEmpty()) {
            sql.append("WHERE ").append(dataFlow.getFilterCondition());
        }

        return sql.toString();
    }

}